package geektime.spring.reactor.simplereactordemo;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

@SpringBootApplication
@Slf4j
public class SimpleReactorDemoApplication implements ApplicationRunner {

  public static void main(String[] args) {
    SpringApplication.run(SimpleReactorDemoApplication.class, args);
  }


  @Override
  public void run(ApplicationArguments args) throws Exception {
    Flux.range(1, 6)
        .publishOn(Schedulers.newElastic("newElastic"))//注意顺序造成的区别
        .doOnRequest(n -> log.info("doOnRequest-Request {} number", n))
        .doOnComplete(() -> log.info("doOnComplete-Publisher COMPLETE 1"))
        .map(i -> {
          log.info("map-Publish {},{}", Thread.currentThread(), i);
              return 10 / (i - 3);
//              return i;
        })
        .doOnComplete(() -> log.info("doOnComplete-Publisher COMPLETE 2"))
        .subscribeOn(Schedulers.newSingle("newSingle"))
        .onErrorResume(e -> {
          log.error("Exception {}", e.toString());
          return Mono.just(-1);
        })
//    .onErrorReturn(-1)
        .subscribe(i -> log.info("subscribe-Subscribe {}:{}", Thread.currentThread(), i),
            e -> log.error("subscribe-error {}", e.toString()),
            () -> log.info("subscribe-Subscriber COMPLETE")
        ,s->s.request(4)
        );
    Thread.sleep(2000);
  }
}
